package greate.project.cron;

import org.apache.camel.AggregationStrategy;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

public class CronApp02 {

    public static void main(String[] args) throws Exception {

        // 创建默认上下文
        CamelContext context = new DefaultCamelContext();

        // 现开启想表达的是 ， camel 在开启后可以动态的新增和关闭路由。
        context.start();

        // 新增路由
        context.addRoutes(new HelloWorldRoute());

        synchronized (CronApp02.class){
            CronApp02.class.wait();
        }

    }

    static class HelloWorldRoute extends RouteBuilder {
        @Override
        public void configure() throws Exception {

            int[] n = {0};

            // cron 表达式使用
            from("cron:tab?schedule=0/1+*+*+*+*+?+*&synchronous=true")
                    .setBody(exchange -> {
                        System.out.println("生产" + ++n[0]);
                        return "1";
                    })
                    .loop(simple("${body}"))
            .to("direct:d1");


            from("direct:d1")
//                    .process(exchange -> {
//                        System.out.println("消费： " + --n[0]);
//                    })
                    .aggregate(constant(true) , (oldExchange, newExchange) -> {

                        if (oldExchange==null) {
                            return newExchange;
                        }

                        final Integer oldBody = oldExchange.getMessage().getBody(int.class);
                        final Integer newBody = newExchange.getMessage().getBody(int.class);
                        oldExchange.getMessage().setBody(oldBody + newBody);
                        return oldExchange;
                    })
                    .completionSize(100)
            .log("消费: ${body}");

        }
    }

}
